package amq.demo;

import java.io.File;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Level;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import lang.JLTool;
import ntnu.eval.rules.IRule;
import ntnu.eval.util.RuleAgent;

import org.apache.activemq.ActiveMQConnectionFactory;

import amq.demo.thds.SQThd;
import amq.utils.GlobalAMQ;
import amq.utils.enums.ELangCode;
import ckip.utils.CKIParser;
import ckip.utils.Node;
import flib.util.Tuple;
import flib.util.io.QSReader;

public class SentConsumer4 implements Runnable{
	public boolean isStop = false;
	String brokerURI;
	String topic;
	JLTool jlt;
	ConnectionFactory connectionFactory;  
    Connection connection = null;  
    Session session = null;
    SQThd sqThd = null;
    String ruleFilePath = null;
    RuleAgent ruleAgent = null;
    CKIParser ckipPsr = null;
	
	public SentConsumer4(String bui, String topic, String rfp)
	{
		this.brokerURI = bui;
		this.topic = topic;
		this.ruleFilePath = rfp;
	}
	
	protected void sayHi()
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(GlobalAMQ.SayHiTopic);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = session.createTextMessage(String.format("%s:hi", this.getClass().getName()));	  
	        message.setStringProperty("Topic", topic);
			producer.send(message);
			System.out.printf("\t[%s] Say Hi...\n", this.getClass().getName());
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	protected void sayBye()
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(GlobalAMQ.SayHiTopic);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = session.createTextMessage(String.format("%s:bye", this.getClass().getName()));
	        message.setStringProperty("Topic", topic);
			producer.send(message);
			System.out.printf("\t[%s] Say Bye...\n", this.getClass().getName());
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	protected void _response(String dest, String tid, String reason)
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(dest);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = null;
	        
			// tuple(Rule_ID,Category,Hit_Part,Correction, IRule)
	        message = session.createTextMessage(reason);
			message.setBooleanProperty("Correct", true);
			message.setLongProperty("TimeStamp", new Date().getTime());
			message.setStringProperty("ServiceName", "AlwaysPass");
			message.setStringProperty("TID", tid);
			producer.send(message);
			System.out.printf("\t[%s] Send answer back to Queue:'%s'!\n", this.getClass().getName(), dest);
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	protected void _response(List<IRule> rtnList, String dest, String tid)
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(dest);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = null;
	        
			// tuple(Rule_ID,Category,Hit_Part,Correction, IRule)
			if(rtnList.size()>0)
			{
				message = session.createTextMessage("Fail");
				message.setBooleanProperty("Correct", false);
				message.setLongProperty("TimeStamp", new Date().getTime());
				message.setStringProperty("TID", tid);
				message.setStringProperty("ServiceName", this.getClass().getName());
				IRule r = rtnList.get(0);
				message.setStringProperty("Category", r.getCategory());			
				message.setStringProperty("Reason", r.getID());
				message.setStringProperty("Suggestion", r.showCorrection());
			}
			else
			{
				message = session.createTextMessage("Pass");
				message.setBooleanProperty("Correct", true);
				message.setLongProperty("TimeStamp", new Date().getTime());
				message.setStringProperty("ServiceName", this.getClass().getName());
				message.setStringProperty("TID", tid);
			}
			producer.send(message);
			System.out.printf("\t[%s] Send answer back to Queue:'%s'!\n", this.getClass().getName(), destination);
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	@Override
	public void run() {				  
        Destination destination;  
        MessageConsumer consumer = null;  
        TextMessage message = null;  
        boolean useTransaction = false;  
        try {       
        	// 0) Start Service Query Thread
        	sqThd = new SQThd(brokerURI, this.getClass().getName(), topic, ELangCode.TraditionChinese);
        	new Thread(sqThd).start();
        	
        	// 1) Prepare RuleAgent
        	ckipPsr = new CKIParser("john", "john7810", null, 2013, Level.WARNING);
        	ruleAgent = new RuleAgent();
        	HashMap<String,IRule> ruleMap = new HashMap<String,IRule>();
        	QSReader qsr = new QSReader(new File(ruleFilePath));
        	qsr.skipCommentLine = qsr.skipEmptyLine = true;
        	qsr.open();
        	int id=1;
        	for(String line:qsr)
        	{
        		String rid = String.format("R%d", id);
        		IRule rule = RuleAgent.RuleStrTranslate(rid, line);
        		if(rule!=null) ruleMap.put(rid, rule);
        		id++;
        	}
        	qsr.close();
        	System.out.printf("\t[Info] Total %d rules...\n", ruleMap.size());
        	ruleAgent.addRules(ruleMap);
        	
        	// 2) Start AMQ Producer        	
            connectionFactory = new ActiveMQConnectionFactory(brokerURI);  
            connection = connectionFactory.createConnection();  
            connection.start();  
            session = connection.createSession(useTransaction,  
                    Session.AUTO_ACKNOWLEDGE);  
            destination = session.createTopic(topic);
            consumer = session.createConsumer(destination);  
            System.out.printf("\t[%s] Consumer starting...\n", this.getClass().getName());
            sayHi();
            while(!isStop)
            {
            	message = (TextMessage) consumer.receive(1000);  
                if(message!=null) {
                	String line = message.getText();
                	System.out.printf("\t[%s] Received message:\n%s\n", this.getClass().getName(), message);                	
                	System.out.printf("\t\tUID: %s\n", message.getStringProperty("UID"));
                	System.out.printf("\t\tTID: %s\n", message.getStringProperty("TID"));
                	System.out.printf("\t\tReplyTo: %s\n", message.getStringProperty("ReplyTo"));
                	System.out.printf("\t\tSENT: %s\n", line);                
                	
                	// Parsing Sentence
                	StringBuffer lineWithPOSBuf = new StringBuffer();
                	Tuple tuple = ckipPsr.queryWithNodeRtn(line);
                	if(((Integer)tuple.get(0))==0)
                	{
                		Node node = (Node)tuple.get(2);
                		Iterator<Node> iter = node.iteratorInNode();
                		while(iter.hasNext())
                		{
                			Node sn = iter.next();
                			lineWithPOSBuf.append(String.format("%s:%s ", sn.label, sn.token));
                		}
                		System.out.printf("\t\tPARSED SENT: %s\n", lineWithPOSBuf.toString().trim());
                		List<IRule> hitRules = ruleAgent.evaluate(lineWithPOSBuf.toString().trim());
                		System.out.printf("\t\tMATCH RULE: %d\n", hitRules.size());
                		_response(hitRules, message.getStringProperty("ReplyTo"), message.getStringProperty("TID"));
                	}
                	else _response(message.getStringProperty("ReplyTo"), message.getStringProperty("TID"), "Parsing Fail");
                }
                
            }            
        } catch (JMSException jmsEx) {  
        	jmsEx.printStackTrace();
        } catch(Exception e){
        	e.printStackTrace();
        } finally {  
            try{
            	sayBye();
            	consumer.close();  
                session.close();  
                connection.close();  
                sqThd.isStop = true;
                sqThd = null;
            } catch(Exception e){}
        }  
        System.out.printf("\t[%s] Bye!\n", this.getClass().getName());        
	}

	
	/**
	 * BD: Sentence Topic Consumer2 - Always response pass
	 * @param args
	 */
	public static void main(String[] args) throws Exception{
		if(args.length<3)
		{
			System.out.printf("\t[Info] Please give\n\tArg1 as Connector URI (ex 'tcp://localhost:61616')" +
					                              "\n\tArg2 as Listening Topic (ex 'NTNU.SENT')" +
					                              "\n\tArg3 as Rule File Path.");
			return;
		}
		SentConsumer4 sc4 = new SentConsumer4(args[0], args[1], args[2]);
		new Thread(sc4).start();
		
		File stop = new File("STOP");
	    while(true)
	    {
	    	if(stop.exists()) break;
	    	Thread.sleep(5000);
	    }
		sc4.isStop = true;
		System.out.printf("\t[Info] Exit %s...\n", sc4.getClass().getName());
	}
}
